package com.flowcloud.kafka.util;

import lombok.SneakyThrows;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Collections;
import java.util.concurrent.ExecutionException;

/**
 * @Author:huangbingbing
 * @Date：Created on 2020/8/23 19:02
 * Description:
 */
public class KafkaAdminManager {
    @Autowired
    private AdminClient adminClient;

	/**
	 * 创建topic
	 *
	 * @param name
	 * @return
	 */
	@SneakyThrows
	public Void createTopic(String name, int numPartitions, short replicationFactor) {
		NewTopic topic = new NewTopic(name, numPartitions, (short) replicationFactor);
		return adminClient.createTopics(Collections.singletonList(topic)).all().get();
	}

	/**
	 * 查询topic
	 *
	 * @param name
	 * @return
	 */
	public String queryTopic(String name) {
		DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(name));
		StringBuffer sb = new StringBuffer("topic信息:");
		try {
			result.all().get().forEach((k, v) -> sb.append("key").append(k).append(";v:").append(v));
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
		}
		return sb.toString();
    }

    /**
     * 删除topic
     * @param topicName
     * @return
     */
    public String deleteTopic(String topicName) {
		adminClient.deleteTopics(Collections.singletonList(topicName));
        return topicName;
    }
}
